#include <Comm/Include/SyncQueue.h>

#include <Comm/Include/LockableObject.h>

SyncQueue::SyncQueue(void)
{
    this->semaphore = new Semaphore();
    this->async_recv_func = 0;
    this->asyn_func_user_data = 0;
    // attributes that would be used in the destructor to wait till the queue is empty
    this->fin_cerrada = false;
    this->fin_sem = new Semaphore();
    this->releasing_thread_on_pop = false;
}

SyncQueue::~SyncQueue(void)
{
    close();
    if (!(this->cola.empty()))
    {
        closeAndDiscardAll();
    }

    delete this->fin_sem;
    delete this->semaphore;
}

void SyncQueue::closeAndWaitTillEmpty()
{
    this->close();
    if (!(this->cola.empty()))
    {
        this->fin_sem->wait();
    }    
}

void SyncQueue::closeAndDiscardAll()
{
    this->close();
    this->releasing_thread_on_pop = true;
    int size = this->cola.size();
    for (int i = 0; i < size; i++)
    {
        SimpleMessage* msg = this->cola.front();
        this->cola.pop();
        delete msg;
    }
}

void SyncQueue::close()
{
    if (this->fin_cerrada)
    {
        return;
    }

    this->fin_cerrada = true;

    if (this->cola.empty())
    {
        this->releasing_thread_on_pop = true;
        this->semaphore->signal();
    }
}

void SyncQueue::push(SimpleMessage* msg)
{
    LockableObject* lock_obj;
    // validations
    if (msg == NULL)
    {
        return;
    }
    if (this->fin_cerrada)
    {
        throw SyncQueueExceptionType::QUEUE_ALREADY_CLOSED;
    }

    lock_obj = new LockableObject(this);
    lock_obj->lock();
    
    // if it has an async receive function, we give the message to it, by calling it
    if (this->async_recv_func != 0)
    {
        this->async_recv_func(msg, this->asyn_func_user_data);
    }
    else
    {
        this->cola.push(msg);
        this->semaphore->signal();
    }

    lock_obj->unlock();
    delete lock_obj;
}

SimpleMessage* SyncQueue::pop(unsigned int timeout)
{
    LockableObject* lock_obj;
    SimpleMessage* msg = NULL;
    if (this->releasing_thread_on_pop)
    {
        throw SyncQueueExceptionType::QUEUE_ALREADY_CLOSED;
    }

    try
    {
        this->semaphore->wait(timeout);
    }
    catch(Semaphore::SemaphoreExceptionType e)
    {
        if (Semaphore::SemaphoreExceptionType::SEMAPHORE_TIMEOUT_EXCEPTION == e)
        {
            throw SyncQueueExceptionType::POP_TIMEOUT;
        }
        else
        {
            //throw -1;
            throw SyncQueueExceptionType::QUEUE_ALREADY_CLOSED;            
        }        
    }    

    if (this->releasing_thread_on_pop)
    {
        throw SyncQueueExceptionType::QUEUE_ALREADY_CLOSED;
    }

    lock_obj = new LockableObject(this);
    lock_obj->lock();
    msg = this->cola.front();
    this->cola.pop();
    lock_obj->unlock();
    
    if ( (this->fin_cerrada) && (this->cola.empty()) )
    {
        this->fin_sem->signal();
        this->releasing_thread_on_pop = true;
    }
    delete lock_obj;
    return msg;
}

SimpleMessage* SyncQueue::front(unsigned int timeout)
{
    LockableObject* lock_obj;
    SimpleMessage* msg = NULL;
    if (this->releasing_thread_on_pop)
    {
        throw SyncQueueExceptionType::QUEUE_ALREADY_CLOSED;
    }
    
    try
    {
        this->semaphore->wait(timeout);
    }
    catch (Semaphore::SemaphoreExceptionType e)
    {
        throw SyncQueueExceptionType::FRONT_TIMEOUT;
    }
    
    if (this->releasing_thread_on_pop)
    {
        throw SyncQueueExceptionType::QUEUE_ALREADY_CLOSED;
    }

    lock_obj = new LockableObject(this);
    lock_obj->lock();
    msg = this->cola.front();
    lock_obj->unlock();
    // as i don't take it, i signal it again
    this->semaphore->signal();
    if ( (this->fin_cerrada) && (this->cola.empty()) )
    {
        this->fin_sem->signal();
        this->releasing_thread_on_pop =  true;
    }

    delete lock_obj;
    return msg;
}

void SyncQueue::installAsyncReceiveFunction(TCallbackAsyncReceiveFunction f, void* user_data)
{
    LockableObject* lock_obj = new LockableObject(this);
    lock_obj->lock();
    
    this->async_recv_func = f;
    this->asyn_func_user_data = user_data;
    
    // give all previous messages contained to the async function
    while ( !(this->cola.empty()) )
    {
        SimpleMessage* msg = this->cola.front();
        this->async_recv_func( msg , this->asyn_func_user_data );
        this->cola.pop();        
    }
    
    lock_obj->unlock();
    delete lock_obj;
}

void* SyncQueue::uninstallAsyncReceiveFunction()
{
    LockableObject* lock_obj = new LockableObject(this);
    lock_obj->lock();

    void* old_func_user_data = this->asyn_func_user_data;

    this->asyn_func_user_data = 0;
    this->async_recv_func = 0;

    lock_obj->unlock();
    delete lock_obj;

    return old_func_user_data;
}
